package com.csdn.eval.spring.mqtt.core;

import com.csdn.eval.spring.mqtt.constants.ChannelName;
import com.csdn.eval.spring.mqtt.constants.Common;
import com.csdn.eval.spring.mqtt.model.MqttEventAnnotationDefinition;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.CollectionUtils;

import java.util.Map;

@Slf4j
public class MqttSubscriber<T> {

    @Autowired
    @Qualifier("AnnotationDispatcherMessageHandler")
    private MessageHandlerService<T> messageHandlerService;

    @Autowired
    @Qualifier("DefaultDispatcherMessageHandler")
    private MessageHandlerService<T> defaultMessageHandlerService;

    @Autowired
    private MqttEventFactory mqttEventFactory;

    @Autowired
    private TopicSubscribeService topicSubscribeService;

    // 处理从MQTT接收的消息
    @ServiceActivator(inputChannel = ChannelName.INBOUND)
    public void handleMessage(Message<?> message) {
        MessageHeaders headers = message.getHeaders();
        String topic = headers.get("mqtt_receivedTopic").toString();
        Map<String, String> topicsMap = topicSubscribeService.subscribeTopics();
        if (!CollectionUtils.isEmpty(topicsMap) && !topicsMap.containsKey(topic)) {
            return;
        }
        // 1. 获取 payload（byte[] 格式）
        byte[] payloadBytes = (byte[]) message.getPayload();
        try {
            // 注解处理
            processAnnotations(topic, payloadBytes, headers);
            // 实现类处理
            processImplementation(topic, payloadBytes, headers);
        } catch (Exception e) {
            log.warn("Failed to deserialize payload for topic {}: {}", topic, e.getMessage(), e);
        }
    }

    private void processAnnotations(String topic, byte[] payloadBytes, MessageHeaders headers) throws Exception {
        // 2. 反序列化为原始对象（如 MyRequest）
        ObjectMapper objectMapper = Common.getObjectMapper();
        MqttEventAnnotationDefinition eventDefinition = mqttEventFactory.getEventDefinition(topic);
        Class<?> classTypeByTopic = mqttEventFactory.getClassTypeByTopic(topic);
        if (classTypeByTopic == null || Object.class.equals(classTypeByTopic)) {
            log.debug("No class type found for topic: {},No implementation of @MqttEventListener", topic);
            return;
        }
        T payload = (T) objectMapper.readValue(payloadBytes, classTypeByTopic);
        messageHandlerService.doHandleMessage(topic, payload, headers, eventDefinition);
    }

    private void processImplementation(String topic, byte[] payloadBytes, MessageHeaders headers) {
        defaultMessageHandlerService.doHandleMessage(topic, payloadBytes, headers);
    }

}